package cn.syncrond.tcm.core.pool;

import java.util.ArrayList;
import java.util.List;
import java.util.Vector;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import cn.syncrond.tcm.core.definition.ITask;
import cn.syncrond.tcm.core.definition.ITaskExceptionResolver;
import cn.syncrond.tcm.core.definition.TaskException;
import cn.syncrond.tcm.core.definition.TaskExecutionStatus;
import cn.syncrond.tcm.core.ext.ITaskPoolEventListener;
import cn.syncrond.tcm.core.ext.TaskPoolEvent;
import cn.syncrond.tcm.core.pool.TaskPool.FutureTask;
import cn.syncrond.tcm.service.log.ILogService;
import cn.syncrond.tcm.service.log.LogLevel;

import com.google.common.util.concurrent.ThreadFactoryBuilder;

/**
 * @Description: 任务相关的线程池执行器
 * 	主要作用是屏蔽原有的提交方式，提供新的基于任务的提交方法
 * @author meepai
 * @date 2016-11-21
 */
public class TaskThreadPoolExecutor extends PausableThreadPoolExecutor {
	
	/**
	 * 日志服务  由外部配置注入
	 */
	private ILogService logService;
	
	/**
	 * 任务异常处理器   由外部配置注入
	 */
	private ITaskExceptionResolver exceptionResolver;
	
	/**
	 * 任务列表的上限值   由外部配置注入
	 */
	private int currentTaskUpperLimit, errorTaskUpperLimit, completeTaskUpperLimit, cancelTaskUpperLimit;
	
	/**
	 * 监听器列表
	 */
	private final List<ITaskPoolEventListener> listeners = new ArrayList<ITaskPoolEventListener>();
	
	/**
	 * 当前任务列表
	 */
	private final List<ITask> currentTasks = new Vector<ITask>();
	
	/**
	 * 已完成的任务列表
	 */
	private final List<ITask> completeTasks = new Vector<ITask>();
	
	/**
	 * 出错的任务列表
	 */
	private final List<ITask> errorTasks = new Vector<ITask>();
	
	/**
	 * 取消的任务列表
	 */
	private final List<ITask> cancelTasks = new Vector<ITask>();
	
	/**
	 * 任务绑定到线程 
	 */
	private final ThreadLocal<ITask> taskLocal = new ThreadLocal<ITask>();
	
	public TaskThreadPoolExecutor(int nThreads){
		super(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), 
				new ThreadFactoryBuilder().setDaemon(true).setNameFormat("TaskThread-%d").build());
	}
	
	public TaskThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
			long keepAliveTime, TimeUnit unit,
			BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
	}

	public TaskThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
			long keepAliveTime, TimeUnit unit,
			BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
				threadFactory);
	}

	public TaskThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
			long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
	}

	public TaskThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
			long keepAliveTime, TimeUnit unit,
			BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
			RejectedExecutionHandler handler) {
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
				threadFactory, handler);
	}
	
	/**
	 * 每个任务执行之前，
	 */
	@Override
	protected void beforeExecute(Thread t, Runnable r) {
		// 在每次执行任务前加锁
		lock.lock();
		log(LogLevel.DEBUG, t.getName() + " 开始执行任务!");
		try {
			if(r instanceof FutureTask){
				// 获取任务
				ITask task = ((FutureTask)r).getTask();
				// 设置等待状态
				task.setExecutionStatus(TaskExecutionStatus.WAITING);
				// while是为了解决多个线程堵塞的时候每个线程都需要重新判断执行条件
				// 增加暂停和当前任务上限限制
				while(isPaused || currentTasks.size() >= currentTaskUpperLimit){
					unpaused.await();
				}
				// 设置任务开始时间
				task.setStartTime(System.currentTimeMillis());
				// 将任务绑定到线程
				taskLocal.set(task);
				// 放入正在执行的任务列表
				currentTasks.add(task);
				// 触发任务开始事件
				task.fireTaskStartEvent();
				// 设置运行状态
				task.setExecutionStatus(TaskExecutionStatus.RUNNING);
			}
		} catch (InterruptedException e) {
			// 如果线程池所在线程中断  那么子任务进程也要中断
			t.interrupt();
			log(LogLevel.ERROR, "任务线程池进程中断，子任务进程也中断！");
		} finally {
			// 执行完释放所
			lock.unlock();
		}
	}
	
	@Override
	@SuppressWarnings("rawtypes")
	protected void afterExecute(Runnable r, Throwable t) {
		// 从线程中获取绑定的任务对象
		ITask task = taskLocal.get();
		// 完成该任务的总耗时
		if(task != null){
			log(LogLevel.DEBUG, task.getName() + " 准备结束任务!");
			// 触发任务结束事件
			task.fireTaskEndEvent();
			// 获取异常
			if(t == null && r instanceof Future){
				t = fetchThrowable((Future)r);
			}
			// 将任务从正在运行列表中移除
			currentTasks.remove(task);
			if(t != null){
				// 触发任务出错事件
				task.fireTaskErrorEvent();
				// 如果是中断异常  则认为是人为取消任务
				if(t instanceof InterruptedException){
					while(cancelTasks.size() >= cancelTaskUpperLimit){
						cancelTasks.remove(0);
					}
					cancelTasks.add(task);
				} else {
					while(errorTasks.size() >= errorTaskUpperLimit){
						errorTasks.remove(0);
					}
					errorTasks.add(task);
				}
				// 捕获异常
				resolveThrowable(t, task);
			} else {
				// 完成任务计数
				task.setEndTime(System.currentTimeMillis());
				while(completeTasks.size() >= completeTaskUpperLimit){
					completeTasks.remove(0);
				}
				completeTasks.add(task);
			}
			// 设置完成状态
			task.setExecutionStatus(TaskExecutionStatus.COMPLETE);
		} else {
			log(LogLevel.ERROR, "有问题，任务不应该从线程绑定对象中消失！");
		}
	}
	
	/**
	 * 获取作业执行中的异常信息
	 * @return
	 */
	private Throwable fetchThrowable(Future<?> future){
		try {
			if(future.isDone()){
				future.get();
			}
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
			return e;
		} catch (ExecutionException e) {
			return e;
		} catch (CancellationException e){
			return e;
		}
		return null;
	}
	
	/**
	 * 捕获异常做不同的处理
	 * @param t
	 */
	private void resolveThrowable(Throwable t, ITask task) {
		log(LogLevel.ERROR, "任务出错, 导致结束!：" + t.getMessage());
		// 任务中断
		if(t instanceof InterruptedException){
			// 如果发生中断，则认为任务被取消
			log(LogLevel.ERROR, t.getMessage());
		} else if(t instanceof ExecutionException){
			Throwable cause = t.getCause();
			if(cause instanceof TaskException){
				if(exceptionResolver != null){
					exceptionResolver.resolve((TaskException)cause);
				}
			} else {
				log(LogLevel.ERROR, cause.getMessage());
			}
		} else if(t instanceof CancellationException){ // 用户取消
			log(LogLevel.ERROR, "用户取消了该作业的执行! [" + task.getName() + "]");
		} else {
			log(LogLevel.ERROR, t.getMessage());
		}
	}
	
	/**
	 * 获取当前任务列表
	 * @return
	 */
	public List<ITask> getCurrentTasks() {
		return currentTasks;
	}

	/**
	 * 获取已经完成的任务列表
	 * @return
	 */
	public List<ITask> getCompleteTasks() {
		return completeTasks;
	}

	/**
	 * 获取出错的任务列表
	 * @return
	 */
	public List<ITask> getErrorTasks() {
		return errorTasks;
	}
	
	/**
	 * 获取取消的任务列表
	 * @return
	 */
	public List<ITask> getCancelTasks() {
		return cancelTasks;
	}

	/**
	 * 添加任务池事件监听
	 * @param listener
	 */
	public void addTaskPoolEventListener(ITaskPoolEventListener listener) {
		listeners.add(listener);
	}

	/**
	 * 触发任务池事件监听
	 * @param event
	 */
	public void fireTaskPoolEvent(TaskPoolEvent event) {
		for(int i=0, n=listeners.size(); i<n; i++){
			listeners.get(i).handleEvent(event);
		}
	}

	/**
	 * 设置日志服务
	 * @param logService
	 */
	public void setLogService(ILogService logService) {
		this.logService = logService;
	}
	
	/**
	 * 设置任务异常处理器
	 * @param exceptionResolver
	 */
	public void setExceptionResolver(ITaskExceptionResolver exceptionResolver) {
		this.exceptionResolver = exceptionResolver;
	}
	
	/**
	 * 获取任务列表的上限
	 * @return
	 */
	public int getCurrentTaskUpperLimit() {
		return currentTaskUpperLimit;
	}

	/**
	 * 设置当前任务列表的上限
	 * @param currentTaskUpperLimit
	 */
	public void setCurrentTaskUpperLimit(int currentTaskUpperLimit) {
		this.currentTaskUpperLimit = currentTaskUpperLimit;
	}

	/**
	 * 获取出错任务列表的上限
	 * @return
	 */
	public int getErrorTaskUpperLimit() {
		return errorTaskUpperLimit;
	}

	/**
	 * 设置出错任务列表的上限
	 * @param errorTaskUpperLimit
	 */
	public void setErrorTaskUpperLimit(int errorTaskUpperLimit) {
		this.errorTaskUpperLimit = errorTaskUpperLimit;
	}

	/**
	 * 获取完成任务列表的上限
	 * @return
	 */
	public int getCompleteTaskUpperLimit() {
		return completeTaskUpperLimit;
	}

	/**
	 * 设置完成任务列表的上限
	 * @param completeTaskUpperLimit
	 */
	public void setCompleteTaskUpperLimit(int completeTaskUpperLimit) {
		this.completeTaskUpperLimit = completeTaskUpperLimit;
	}

	/**
	 * 记录日志
	 * @param level
	 * @param message
	 */
	public void log(LogLevel level, String message) {
		if(logService != null){
			logService.log(level, message);
		}
	}
}
